package core;

import fundamental.Region;
import fundamental.Task;
import hbase.HBaseUtil;
import log.WriteToLog;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.util.Bytes;
import singleobject.MemoryInfo;


import java.io.IOException;
import java.util.*;

/**
 * @Author zhangxin
 * @Date 2017-07-05 16:21:45
 * @Desciption region移动相关操作
 **/
public class MoveOrNot {
    private double threshold = 0.7;
    private Region region = MemoryInfo.getREGION();
    private Task task = MemoryInfo.getTASK();
    private Double requestQueueSize = 5.0;
    static Connection conn = HBaseUtil.getHBASECONFIG().getConnection();

    /**
     * @Author zhangxin
     * @Date 2017-07-05 16:38:41
     * @Desciption 主程序
     **/
    public void core(String taskId,String regionEncodeName){
        //Region region = new Region();
      //  Task task = new Task();
        //1、计算region使用量
        List<String> taskList = region.getTheRegionTask(regionEncodeName);
        Integer regionTaskNum = taskList.size();
        //2、region使用量不为0
        String taskRSId = task.getTheTaskRS(taskId);  //当前task使用的RS
        WriteToLog.write("core code is starting,the task: "+taskId+" is assign to RS: "+taskRSId+" by default");
        if(regionTaskNum != 1){                       //除去当前task，region还被别的task使用
            WriteToLog.write("region is used!!!!!!!!!");
            //2.1 计算移动和不移动region的数据本地性
            String moveToRSId = getMovedRSwhenRegionisUsed(taskList);
            //2.2 判断是否移动，若目的RS与当前RS不同，则移动
            if(!taskRSId.equals(moveToRSId)){
                WriteToLog.write("region is used by other task,move it to the min load RS, so move from "+taskRSId+" to "+moveToRSId);
                moveRegion(regionEncodeName,moveToRSId);
            }
        }
        //3、region使用量为0
        else{
            WriteToLog.write("region is not used, then calculate dest RS load");
            double load = calLoad(taskRSId);
            WriteToLog.write("dest RS load is: "+load);
            //3.1 目的机器负载小于阀值
            if(load < threshold){
                WriteToLog.write("the dest machine can load the region,so move!!!!!!!!!!!!!!!!!!!!!!");
                moveRegion(regionEncodeName,taskRSId);
            }
            //3.2 目的机器负载大于阀值
            else{
                WriteToLog.write("the dest machine cannot load the region");
                String regionRSId = region.getTheRegionRS(regionEncodeName);
                load = calLoad(regionRSId);
                if(load > threshold){
                    WriteToLog.write("move the region to the best machine");
                    moveRegion(regionEncodeName,calMinLoad());
                }
            }
        }
    }


    /**
     * @Author zhangxin
     * @Date 2017-07-07 16:33:25
     * @Desciption region的使用量不为0，寻找最能够满足数据本地性的RS并移动，若存在多个数据本地性相同的RS，则选择请求次数最少的RS
     **/
    public String getMovedRSwhenRegionisUsed(List<String> taskIdList) {
        Map<String,Integer> rsLocality = new HashMap<String, Integer>();
        List<String> taskNumEqualRS = new ArrayList<String>();
        int taskNum = 0;
        int flag = 0;
        String moveToRSId = "";
        for(String taskId : taskIdList){
            String rsId = task.getTheTaskRS(taskId);
            WriteToLog.write("task id and it's dest RS is:"+taskId+" "+rsId);
            if(rsLocality.get(rsId) == null){
                rsLocality.put(rsId,1);
            }
            else{
                taskNum = rsLocality.get(rsId);
                taskNum ++;
                rsLocality.put(rsId,taskNum);
            }
        }
        Iterator<String> key = rsLocality.keySet().iterator();
        while(key.hasNext()){
            String rsId = key.next();
            Integer taskNumforRS = rsLocality.get(rsId);
            WriteToLog.write("in getMovedRSWhenRegionIsUsed() RS id: "+rsId+" task num: "+taskNumforRS);
            if(flag == taskNumforRS){
                taskNumEqualRS.add(rsId);
            }
            if(flag < taskNumforRS){
                taskNumEqualRS.clear();
                moveToRSId = rsId;
                flag = taskNumforRS;
                taskNumEqualRS.add(moveToRSId);
            }
            for(String s : taskNumEqualRS) {
                WriteToLog.write("in getMovedRSWhenRegionIsUsed() task num equal RS id: " + s);
            }
        }
        if(taskNumEqualRS.size() > 1){
            moveToRSId = calMinLoad(taskNumEqualRS);
        }
        WriteToLog.write("in getMovedRSWhenRegionIsUsed() min load RS: "+moveToRSId);
        return moveToRSId;
    }
    /**
     * @Author zhangxin
     * @Date 2017-07-05 16:23:15
     * @Desciption 移动region,并更新信息！！！
     **/
    public void moveRegion(String regionEncodeName, String rsId){
        System.out.println("region: !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"+regionEncodeName);
        try {
            Admin admin = conn.getAdmin();
            admin.move(Bytes.toBytes(regionEncodeName.split("[.]")[1]),Bytes.toBytes(rsId));   //这里的rsId应该是host+port+startcode
            region.addRegionRS(regionEncodeName,rsId);

        } catch (IOException e) {

            e.printStackTrace();
        }
    }

    /**
     * @Author zhangxin
     * @Date 2017-07-05 16:22:56
     * @Desciption 计算机器当前负载，即RS的请求次数的均值与集群请求次数的均值之比
     **/
    public double calLoad(String rsId){
        Map<String, Queue<Integer>> requestMap;
        synchronized (region) {
            requestMap = region.getRequestTimesQueue();
            for(String id : requestMap.keySet())
                System.out.println("计算集群负载时调用  "+id + "    " + requestMap.get(id));
        }
        Queue<Integer> requestQueue = requestMap.get(rsId);
        Double ave = 0.0;
        Double aveToal = 0.0;
        Double loadCoe = 0.0;
        Integer request;
        Double totalForOneRS = 0.0;
        Double total = 0.0;
        int queuesizeForOneRS = requestQueue.size();
        for(Integer requestTime : requestQueue){
            totalForOneRS += requestTime;
            WriteToLog.write("in calLoad() RS id and requeset time is: "+rsId+","+requestTime);
        }
        ave = totalForOneRS / queuesizeForOneRS;     //当前RS的平均请求次数
        WriteToLog.write("in calLoad() queuesizeForOneRS: "+queuesizeForOneRS+" totalForOneRS: "+totalForOneRS);
        Iterator<String> it = requestMap.keySet().iterator();
        while(it.hasNext()){
            int totalrequest = 0;
            String id = it.next();
            Queue<Integer> queue = requestMap.get(id);
            int queuesize = queue.size();
            for(Integer requestTime: queue ){
                totalrequest += requestTime;
                WriteToLog.write("cal every RS request,request time is: "+requestTime);
            }
            WriteToLog.write("totalrequest is: "+totalrequest);
            total += Double.valueOf(totalrequest) / queuesize;
            WriteToLog.write("queuesize: "+queuesize+"total: "+total);
        }
        aveToal = total /  requestMap.keySet().size();   //集群的平均请求次数
        if(aveToal != 0.0){
            loadCoe = ave / aveToal;
        }
        WriteToLog.write("the regionServer and its avg request time is "+rsId+" , "+loadCoe+" , "+aveToal+" , "+ave);
        return loadCoe;
    }

    /**
     * @Author zhangxin
     * @Date 2017-07-05 18:41:01
     * @Desciption 计算当前集群中请求次数最小的RS
     **/
    public String calMinLoad(){
        Integer request;                    //每个RS每秒的请求次数
        String minLoadRsId = "";            //request最少的RS的ID
        Double ave = 0.0;                   //每个RS5秒内的平均request次数
        Double flag = Double.MAX_VALUE;    //标志，用来寻找request最少的RS
        Map<String,Queue<Integer>> requestMap;
        synchronized (region) {
            requestMap = region.getRequestTimesQueue();
        }
        Iterator<String> it = requestMap.keySet().iterator();
        while (it.hasNext()){
           String rsId = it.next();
           Double total = 0.0;
           Queue<Integer> queue = requestMap.get(rsId);
           int queuesize = queue.size();
           for(Integer requestTime: queue){
               total += requestTime;
               WriteToLog.write("in calMinLoad() requestTime is: "+requestTime);
           }
           ave = total / queuesize;
           WriteToLog.write("calculate min request, RS "+rsId+" "+ave);
           if(flag > ave){
               minLoadRsId = rsId;
               flag = ave;
           }
        }
        WriteToLog.write("the min load RS id is："+minLoadRsId);
        return minLoadRsId;
    }

    /**
     * @Author zhangxin
     * @Date 2017-07-07 16:24:26
     * @Desciption 根据RS的集合，求请求次数最小的RS
     **/
    public String calMinLoad(List<String> rsIdList){
        Map<String,Queue<Integer>> requestMap;
        synchronized (region){
            requestMap = region.getRequestTimesQueue();
        }
        Double ave = 0.0;                   //每个RS5秒内的平均request次数
        Double flag = Double.MAX_VALUE;    //标志，用来寻找request最少的RS
        Integer request = 0;
        String minLoadRsId = "";            //request最少的RS的ID
        for(String id : rsIdList){
            Double total = 0.0;
            Queue<Integer> queue = requestMap.get(id);
            int queuesize = queue.size();
            for(Integer requestTime: queue){
                total += requestTime;
                WriteToLog.write("in calMinLoad(list) requestTime is: "+requestTime);
            }
            ave = total / queuesize;
            if(flag > ave){
                flag = ave;
                minLoadRsId = id;
            }
        }
        return minLoadRsId;
    }
}
